
                /*********************************************
                **                                          **
                **                                          **
                **                  reduce                  **
                **                                          **
                **                                          **
                *********************************************/
                /*********************************************
                **                                          **
                **         Author Jean-Marc Adamo           **
                **       Copyright (C) 1997-2008 JMA        **
                *********************************************/


/*
** execute with 6 processes
*/

#include "arch.h"
#include "A.utilities"

//user program

Coll_id cid0, cid1, cid2;

//on most homogenous systems, explicit setting 
//won't be necessary (try commenting the
//setting lines, you should get the same values
//for cid0, cid1 and cid2).
void init(int, char **){
   cid0.set();
   cid1.set();
   cid2.set();
   fprintf(Sdf::trace, "node%d: cid0=%d, cid1=%d, cid2=%d\n",
           Sdf::PROC_ID, cid0.get(), cid1.get(), cid2.get());
   fflush(Sdf::trace);
}

class Proc: public S_Thread{

   public:
        int pid;
        int step;

        Proc(int p, int s)
                : S_Thread((Smf0_ptr)&Proc::body){pid = p; step = s;}

        void body(){
               fprintf(Sdf::trace, "node %d, in proc %d  at start %d\n",
                       Sdf::PROC_ID, pid, step);
               fflush(Sdf::trace);
               int i;
               for(i = 0; i < 4; i++){
                   reschedule();
               }

               if(step<1){

                   fprintf(Sdf::trace, "node %d, in proc %d ***before t_reduce at step %d\n",
                           Sdf::PROC_ID, pid, step);
                   fflush(Sdf::trace);

                   Group *newgroup;
                   int rk[6];
                   for(i = 0; i < Sdf::PROC_NB; i++)
                      rk[i] = Sdf::PROC_NB-i-1; //reverse order
                   newgroup = Sdf::incl(Sdf::COMML_WORLD, Sdf::PROC_NB, rk);
                   CommL *newcomm = newgroup->commit(cid0, Sdf::COMML_WORLD);

                   if(pid == 1){
                      int from1[2];
                      from1[0] = pid*10 + Sdf::PROC_ID;
                      from1[1] = pid*50 + Sdf::PROC_ID;

                      int to1[2];
                      to1[0]   = 777;
                      to1[1]   = 777;

                      _Sdf<int>::reduce(from1[0], to1[0], 2,
                                        Opa<int>::SUM, 4, cid1, newcomm);

                      fprintf(Sdf::trace, "node %d, in proc %d ***after t_reduce, from=%d,%d, to =%d,%d\n",
                              Sdf::PROC_ID, pid, from1[0], from1[1], to1[0], to1[1]);
                      fflush(Sdf::trace);

                   }else{

                      MINMAX_pair<int> from2[2];
                      from2[0].set_data(1);
                      from2[0].set_rank(Sdf::PROC_NB-1-Sdf::PROC_ID);
                      from2[1].set_data(1);
                      from2[1].set_rank(Sdf::PROC_ID);

                      int to_t = 777;
                      MINMAX_pair<int> to2[2];
                      to2[0].set_data(777);
                      to2[0].set_rank(777);
                      to2[1].set_data(777);
                      to2[1].set_rank(777);

                      typedef MINMAX_pair<int> My_pair;
                      _Sdf<My_pair>::reduce(from2[0], to2[0], 2, 
                                            Opm<int>::MAXLOC, 2, cid2, newcomm);

                      fprintf(Sdf::trace, "node %d, in proc %d ***after t_reduce, rank=%d, from=%d,%d; %d,%d to =%d,%d; %d,%d\n",
                              Sdf::PROC_ID, pid, rk[Sdf::PROC_ID], 
                              from2[0].get_data(), from2[0].get_rank(), 
                              from2[1].get_data(), from2[1].get_rank(), 
                              to2[0].get_data(), to2[0].get_rank(), 
                              to2[1].get_data(), to2[1].get_rank());
                      fflush(Sdf::trace);
                   }

                   delete newcomm;

                   Proc *proc1 = new Proc(pid*10 + 1, step+1);
                   Proc *proc2 = new Proc(pid*10 + 2, step+1);
                   S_Thread *proc_tabl[2] = {proc1, proc2};
                   int pri[2] = {1, 1};
                   par(2, proc_tabl, pri);
                   delete proc1;
                   delete proc2;
               }
               fprintf(Sdf::trace, "on node %d, proc %d the end at step %d\n",
                        Sdf::PROC_ID, pid, step);
               fflush(Sdf::trace);
        }
};

class Root_proc: public Thread{

   public:

        Root_proc()
                : Thread((Tmf0_ptr)&Root_proc::body){}

        void body(){

                fprintf(Sdf::trace, "node %d, in Root_proc::body1\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                Proc *proc1 = new Proc(1, 0);
                Proc *proc2 = new Proc(2, 0);
                fprintf(Sdf::trace, "node %d, in Root_proc::body2\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                S_Thread *proc_tabl[2] = {proc1, proc2};
                int pri[2] = {1, 1};
                par(2, proc_tabl, pri);
                fprintf(Sdf::trace, "node %d, in Root_proc::body4\n", Sdf::PROC_ID);
                fflush(Sdf::trace);
                delete proc1;
                delete proc2;
        }
};
//end of user program

MAIN(Root_proc, init, Sdf::empty, Sdf::empty)

